package com.learn.kafka.timestamp;

import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;

/**
 * 根据时间戳查找偏移量并消费消息的消费者（继承通用配置父类）
 * @author: learn.kafka
 */
public class TimestampConsumer extends ConsumerProperties {

    public static void main(String[] args) {
        new TimestampConsumer().consumeByTimestamp();
    }

    // 构造方法：初始化时间戳消费的特殊配置
    public TimestampConsumer() {
        super(); // 调用父类构造方法，加载通用配置
        initTimestampConfig();
    }

    // 初始化时间戳消费的特殊配置（覆盖父类默认值）
    private void initTimestampConfig() {
        // 设置消费组ID（覆盖父类默认值）
        setProperty("group.id", "mate-time-group");
        // 取消自动偏移量重置（由时间戳查询决定起始位置）
        setProperty("auto.offset.reset", "none");
    }

    // 核心逻辑：根据时间戳查找偏移量并消费消息
    private void consumeByTimestamp() {
        // 从父类获取消费者实例
        Consumer<Integer, String> consumer = createConsumer();
        String topic = "a4part2repq";

        try {
            // 1. 获取主题的所有分区信息
            List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
            List<TopicPartition> topicPartitions = new ArrayList<>();
            Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();

            // 2. 计算目标时间戳（7天前）
            DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long nowTime = new Date().getTime();
            long fetchDataTime = nowTime - 1000L * 60 * 60 * 24 * 7; // 7天前的时间戳
            System.out.println("查询时间戳（7天前）：" + fetchDataTime + " (" + df.format(new Date(fetchDataTime)) + ")");

            // 3. 为每个分区设置查询时间戳
            for (PartitionInfo partitionInfo : partitionInfos) {
                TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
                topicPartitions.add(topicPartition);
                timestampsToSearch.put(topicPartition, fetchDataTime);
            }

            // 4. 手动分配分区
            consumer.assign(topicPartitions);

            // 5. 根据时间戳获取各分区的偏移量
            Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(timestampsToSearch);
            System.out.println("各分区初始偏移量设置：");
            for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {
                OffsetAndTimestamp offsetTimestamp = entry.getValue();
                if (offsetTimestamp != null) {
                    TopicPartition partition = entry.getKey();
                    long offset = offsetTimestamp.offset();
                    System.out.printf("分区 %d：时间戳=%d，偏移量=%d%n",
                            partition.partition(), offsetTimestamp.timestamp(), offset);
                    // 设置消费者从该偏移量开始消费
                    consumer.seek(partition, offset);
                } else {
                    System.out.printf("分区 %d：在指定时间戳之后无消息%n", entry.getKey().partition());
                }
            }

            // 6. 从指定偏移量开始消费消息
            while (true) {
                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.printf("分区=%d，偏移量=%d，消息内容=%s%n",
                            record.partition(), record.offset(), record.value());
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}
